package com.yanggu.mapreduce.compress;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.io.compress.CompressionInputStream;
import org.apache.hadoop.io.compress.CompressionOutputStream;
import org.apache.hadoop.util.ReflectionUtils;

import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.File;

/**
 * @version V1.0
 * @author: YangGu
 * @date: 2020/2/8 16:44
 * @description:
 */
public class TestCompress {

    public static void main(String[] args) throws Exception {
        compress("input/compress/a.txt", "output/compress/a.txt", "org.apache.hadoop.io.compress.DefaultCodec");
        //decompress("input/compress/a.txt");
    }

    private static void compress(String pathName, String pathOutName, String method) throws Exception {
        //1. 创建输入流
        FileInputStream fis = new FileInputStream(pathName);

        //2. 通过反射创建压缩的编解码器
        Class clazz = Class.forName(method);

        CompressionCodec codec = (CompressionCodec) ReflectionUtils.newInstance(clazz, new Configuration());

        //3. 获取输出流
        FileOutputStream fos = new FileOutputStream(pathName + codec.getDefaultExtension());
        //创建压缩的输出流
        CompressionOutputStream cos = codec.createOutputStream(fos);

        //4. 流的拷贝
        IOUtils.copyBytes(fis, cos, 1024 * 1024, false);

        //5. 关闭流
        IOUtils.closeStream(cos);
        IOUtils.closeStream(fos);
        IOUtils.closeStream(fis);

    }

    // 2、解压缩
    private static void decompress(String filename) throws FileNotFoundException, IOException {

        // （0）校验是否能解压缩
        CompressionCodecFactory factory = new CompressionCodecFactory(new Configuration());

        CompressionCodec codec = factory.getCodec(new Path(filename));

        if (codec == null) {
            System.out.println("cannot find codec for file " + filename);
            return;
        }

        // （1）获取输入流
        CompressionInputStream cis = codec.createInputStream(new FileInputStream(new File(filename)));

        // （2）获取输出流
        FileOutputStream fos = new FileOutputStream(new File(filename + ".decoded"));

        // （3）流的对拷
        IOUtils.copyBytes(cis, fos, 1024*1024*5, false);

        // （4）关闭资源
        cis.close();
        fos.close();
    }


}
